package com.doit.beans.day06

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DataTypes, StructType}

/**
 * @Author:
 * @WX: 17710299606
 * @Tips: 学大数据 ,到多易教育
 * @DOC: https://blog.csdn.net/qq_37933018?spm=1000.2115.3001.5343
 * @Description:
 */
object Demo03_DF_File {
  def main(args: Array[String]): Unit = {

    val session = SparkSession.builder()
      .appName("test")
      .master("local[*]")
      .getOrCreate()


    /*
       准备  ORC  PARQUET
      val structType = new StructType()
         .add("oid", DataTypes.StringType)
         .add("money", DataTypes.DoubleType)
         .add("city", DataTypes.StringType)
         .add("cate", DataTypes.StringType)
         .add("uid", DataTypes.StringType)
       val df = session.read
         .schema(structType)
         .csv("data/orders/order.csv")
       df.createTempView("tb_orders")


       session.sql(
         """
           |select
           |oid ,
           |money ,
           |cate
           |from  tb_orders
           |""".stripMargin).write
           .parquet("data/parquet")
           //.orc("data/orc")
   */

    val df1 = session.read.orc("data/orc")
    df1.show()
    df1.printSchema()
    println("----------------------------------------------")
    val df2 = session.read.parquet("data/parquet")
    df2.show()
    df2.printSchema()

  }

  private def fromCsv02(session: SparkSession): Unit = {

    /**
     * 根据数据自定义 schema   数据结构
     */
    val structType = new StructType()
      .add("oid", DataTypes.StringType)
      .add("money", DataTypes.DoubleType)
      .add("city", DataTypes.StringType)
      .add("cate", DataTypes.StringType)
      .add("uid", DataTypes.StringType)
    val df = session.read
      .schema(structType)
      .csv("data/orders/order.csv")

    df.printSchema()
    df.show()
  }

  private def fromCsv01(session: SparkSession): Unit = {
    /**
     * csv文件中可能没有列属性名
     * root
     * |-- _c0: string (nullable = true)
     * |-- _c1: string (nullable = true)
     * |-- _c2: string (nullable = true)
     * |-- _c3: string (nullable = true)
     * |-- _c4: string (nullable = true)
     *
     * 1) 假如csv文件的第一行是数据列属性名
     * .option("header"  ,"true")  制定第一行数据就是列 --> 字段名
     * root
     * |-- oid: string (nullable = true)
     * |-- price: string (nullable = true)
     * |-- city: string (nullable = true)
     * |-- category: string (nullable = true)
     * |-- uid: string (nullable = true)
     *
     * 2)  根据列的值  推导数据类型
     * .option("inferSchema"  ,"true")
     */
    val df = session.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("data/orders/order.csv")
    df.printSchema()
    df.show()
  }

  private def fromJsonFile(session: SparkSession): Unit = {
    /**
     * 1. json 本身具备数据结构的属性  (属性名 , 数据类型)
     * {"uid":1 , "name":"zss"}
     * 1)  返回的DF中包含json数据中所有的属性  , 行数据没有的属性补充null
     * 2) 当遇到错误行数据 , 解析失败 所有的字段都为null  多出一个额外字段  _corrupt_record
     */
    val jsonDF = session.read
      .json("data/log")

    jsonDF.printSchema() // 打印结构
    jsonDF.show() //  展示数据

    jsonDF.createTempView("a")

    session.sql(
      """
        |select
        |*
        |from
        |a
        | where  _corrupt_record is  null
        |""".stripMargin).show()
  }
}
